package com.bigdata.streaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo3ReduceByKeyAndWindow {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local[2]")
    conf.setAppName("state")
    val sc = new SparkContext(conf)

    sc.setLogLevel("error")

    // 5秒钟一个窗口，5秒钟会触发一次计算
    val ssc = new StreamingContext(sc, Durations.seconds(5))

    val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("node01", 8888)

    val kvDS: DStream[(String, Int)] = linesDS.flatMap(_.split(',')).map((_, 1))

    /**
     * 滑动窗口
     * 每隔5分钟，统计一次最近30分钟的订单量，订单金额（双11）
     * 统计最近15秒单词的数量，每隔5秒统计一次
     *
     * 窗口大小和滑动时间必须是park streaming拉取数据间隔时间的整数倍(5的倍数)
     *
     * hello,word    -> (hello,1) (word,1)
     * hello,hive    -> (hello,2) (word,1) (hive,1)
     * hello,word    -> (hello,3) (word,2) (hive,1)
     * hello,spark   -> (hello,3) (word,1) (hive,1) (spark,1)
     * hello,hbase   -> (hello,3) (word,1) (spark,1) (hbase,1)
     */
    //     val countDS: DStream[(String, Int)] = kvDS.reduceByKeyAndWindow(
    //       (x: Int, y: Int) => x + y, //聚合函数
    //       Durations.seconds(15), //窗口大小
    //       Durations.seconds(5) //滑动时间
    //     )

    /**
     * 滚动窗口  -- 每隔15秒计算一次，2个时间都需要手动指定
     * 滚动窗口是一种特殊的滑动窗口，2个窗口时间没有数据交集
     * 特殊的滑动窗口：窗口大小等于滑动时间
     */
    val countDS: DStream[(String, Int)] = kvDS.reduceByKeyAndWindow(
      (x: Int, y: Int) => x + y, // 聚合函数
      Durations.seconds(5), // 窗口大小
      Durations.seconds(5) // 滑动时间
    )


    countDS.print()

    ssc.start()
    ssc.awaitTermination()
    ssc.stop()


  }

}
